-- drop function if exists sm_rdd.fv_rdd_map_input(varchar(64), text, text, int[]);
create or replace function sm_rdd.fv_rdd_map_input
(
  i_frame          text[]        ,    -- fv_rdd_map_input(i_frame[frame[], i_param...], i_lambda)  ===  aggr(f_lambda(frame[cur] , i_param...))
  i_lambda         text          ,    -- 格式形如 'x text, y int -> (select least(length(x), y))'，其中()内是 select segment
  i_node_lists     int[]              -- null 时，表示所有 node 参与计算，此时 worker 本地数据可能参与计算，不保证幂等性；由 sm_rdd.tb_snap_cluster_node_list.node_status = '3' and is_worker is true 和 i_node_lists 共同保证幂等性
)
returns int
as
$$
declare -- here
  v_job_no              int;

-- fv_get_lambda_param.. 如果使用text[]强制类型转换方法，需要规约很多参数定义代码的边界约束。
-- 例如：select ('{x decimal(32,8)[], y int}'::text[])[1] 参数定义code返回不完整，要select ('{"x decimal(32,8)[]", y int}'::text[])[1]才可以。
-- 这里选择regex做词法解析
  v_lambda_id           char(32)  :=   md5(sm_rdd.fv_get_lambda_param(i_lambda)::text || sm_rdd.fv_get_lambda_code(i_lambda)::text);   
  v_frame               text[]    :=   i_frame[1]::text[];

-- 可选分发策略：
-- 1. node_list 驱动，全部可用节点都执行，适合场景：MPP，固定每个节点参与计算(tb_snap_cluster_node_list.node_status = 3 and is_worker is true)、数据已分布式存储在各节点，从各节点分布式存储获取计算对象（不仅仅包括入参，更包括节点本地数据）
-- 2. 弹性分配集群资源，部分可用节点执行，适合场景：RDD，自行切割入参分发，各节点本地数据不参与计算
  v_node_lists          int[]     := 
  (
    select array_agg(node_id order by node_id) 
    from
    (
      select node_id from sm_rdd.tb_snap_cluster_node_list 
      where (node_status in ('1', '3') and i_node_lists is not null and node_id = any(i_node_lists) 
          or i_node_lists is null and node_status = '1' and last_free_heartbeat_time >= now() - interval '10 second')   -- 十秒内心跳的空闲节点
        and is_worker is true
      limit array_length(v_frame, 1)
    ) t_a_tmp
  )  
  ;
  v_cur_map_strategy    cursor
    for select node_id
          , row_number() over(order by node_id) as node_rn   -- 用户分发策略
        from unnest(v_node_lists) node_id 
  ;


-- declare   v_sql_code            text;
  v_map_no              int       :=   0;
  v_record              record  ;
  v_uuid                varchar(64) :=  (gen_random_uuid())::varchar(64);    -- pg 13 不在需要此扩展，用 gen_random_uuid() 代替 uuid_generate_v4()


begin
  if v_node_lists is null
  then
    raise exception 'no free workers!';
  end if;
  
  -- 算子准备
  if not exists(select  from sm_rdd.tb_snap_lambda where lambda_id = v_lambda_id)
  then
    insert into sm_rdd.tb_snap_lambda
    (
      lambda_id                ,
      -- -- lambda_inner_name        ,
      lambda_params            ,
      lambda_code              ,
      lambda_status
    )
    select 
      v_lambda_id                                    ,
      -- -- 'sm_rdd.fn_lambda_' || v_lambda_id     ,
      sm_rdd.fv_get_lambda_param(i_lambda)::varchar(64)[]     ,
      sm_rdd.fv_get_lambda_code(i_lambda)::text               ,
      '1'
    ;

    -- -- -- 在各个节点生成 lambda
    -- -- perform 
    -- --   sm_rdd.fv_create_snap_master_lambda_v2
    -- --   (
    -- --     v_lambda_id,
    -- --     sm_rdd.fv_get_lambda_param(i_lambda),
    -- --     sm_rdd.fv_get_lambda_code(i_lambda),
    -- --     v_node_lists
    -- --   )
    -- -- ;
  end if;
  
  -- 注册任务
  -- -- select nextval('sm_rdd.tb_trs_distrb_job_job_no_seq') into v_job_no;
  insert into sm_rdd.tb_trs_distrb_job
  (
    -- -- job_lambda_name    ,
    job_lambda_id
  )
  select
    -- -- md5(sm_rdd.fv_get_lambda_param(i_lambda)::text || sm_rdd.fv_get_lambda_code(i_lambda))   ,
    v_lambda_id
  returning job_no into v_job_no
  ;
  
  open v_cur_map_strategy;
  
  loop
    -- 分发任务
    fetch v_cur_map_strategy into v_record;
    exit when not found;

    v_map_no := v_map_no + 1;
    insert into sm_rdd.tb_parted_trs_work_list
    (
      work_node_id            ,
      work_no                 ,
      work_job_no             ,
      work_lambda_id          ,
      work_status             ,
      v_filter
    )
    select 
      v_record.node_id          ,
      nextval('seq_trs_work_list'),
      v_job_no              ,
      v_lambda_id           ,
      'init'                  ,
      v_uuid
    from generate_series(1, array_length(v_frame, 1)) frame_cnt
    where (frame_cnt - 1) % array_length(v_node_lists, 1) + 1 = v_record.node_rn   -- 蛇形分发，比如 a, b, c 三个 worker 节点，1 - 8 个 work 任务，那么分发为：1a, 2b, 3c, 4a, 5b, 6c, 7a, 8b
    -- -- 多条无法用 returing into 语句传递给 job_data 表，而用 v_filter 分组标记 
    ;

    -- 分发数据
    insert into sm_rdd.tb_parted_sour_job_data
    (
      work_no              ,
      work_node_id         ,
      flat_map_key         ,
      work_input_frame
    )
    select 
      work_no              ,
      work_node_id         ,
      array_length(v_node_lists, 1) * ((row_number() over()) - 1) + v_map_no           ,          -- i_frame 切片序号当作 map_key
      v_frame[array_length(v_node_lists, 1) * ((row_number() over()) - 1) + v_map_no] || i_frame[2:]     -- 按照规约，找到该 worker 节点领取 work 的对应多个数据分片
    from sm_rdd.tb_parted_trs_work_list
    where work_node_id = v_record.node_id
      and v_filter = v_uuid
    ;

    -- 各个 worker 节点的 pg_agent 轮询触发任务执行
    -- -- workers are working
    -- -- working

  end loop;
  
  close v_cur_map_strategy;
  
  -- 确定 job 的 job_map_work_cnt
  update sm_rdd.tb_trs_distrb_job
  set job_map_work_cnt = array_length(v_frame, 1)
  where job_no = v_job_no
  ;

  return v_job_no;  
end
$$
language plpgsql volatile
cost 100;

-- select 
--   sm_rdd.fv_rdd_map_input
--   (
--     , i_frame       => array['{"aaa", "bbbbb", "cccc", "ddddddd", "ee"}', '3']    
--     , i_lambda      => 'x text, y int -> select least(length(x), y)'
--     , i_node_lists  => null
--   )
